package cn.topcodes.tcsf.admin.jms;

import cn.topcode.unicorn.utils.CollectionUtil;
import cn.topcode.unicorn.utils.PropertyConfigurer;
import cn.topcode.unicorn.utils.StringUtil;
import cn.topcode.unicorn.utils.dist.DestType;
import cn.topcode.unicorn.utils.dist.DistEventConfig;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import javax.annotation.PostConstruct;
import java.util.*;

/**
 * Created by Unicorn on 2018/5/2.
 */
public class DistEventRegister implements ApplicationContextAware {


    /**
     * 消费的事件
     */
    private Map<String, Class> consume = new HashMap<>();

    /**
     * 生产的事件
     */
    private Map<String, Class> produce = new HashMap<>();

    /**
     * 监听的Queue
     */
    private Set<String> queueDestNames = new HashSet<>();

    /**
     * 监听的Topic
     */
    private Set<String> topicDestNames = new HashSet<>();

    private List<Class> mConsume;

    private List<Class> mProduce;

    @Autowired(required = false)
    private PropertyConfigurer propertyConfigurer;

    private ApplicationContext applicationContext;

    /**
     * 监听的VirtualTopic
     */
    private Set<String> virtualTopicDestNames = new HashSet<>();

    public DistEventRegister(List<Class> consume, List<Class> produce) {
        if(consume.isEmpty() && produce.isEmpty()) {
            throw new IllegalArgumentException("请至少设置一个生产或消费的分布式事件类型");
        }

        this.mConsume = consume;
        this.mProduce = produce;
    }

    @PostConstruct
    public void init() {

        for(Class clazz : mProduce) {
            DistEventConfig config = (DistEventConfig) clazz.getAnnotation(DistEventConfig.class);
            if(config != null) {
                if(StringUtil.isNotBlank(config.type())) {
                    this.produce.put(config.type(), clazz);
                }else {
                    this.produce.put(clazz.getName(), clazz);
                }
            }
        }

        String consumerName = null;
        for(Class clazz : mConsume) {
            DistEventConfig config = (DistEventConfig) clazz.getAnnotation(DistEventConfig.class);
            if(config != null) {
                if(StringUtil.isNotBlank(config.type())) {
                    this.consume.put(config.type(), clazz);
                }else {
                    this.consume.put(clazz.getName(), clazz);
                }

                String destName = config.destName();
                if(StringUtil.isBlank(destName)) {
                    destName = clazz.getName();
                }
                if(DestType.QUEUE.equals(config.destType())) {
                    queueDestNames.add(destName);
                }else if(DestType.TOPIC.equals(config.destType())) {
                    topicDestNames.add(destName);
                }else if(DestType.VIRTUAL_TOPIC.equals(config.destType())) {
                    if(StringUtil.isBlank(consumerName)) {
                        consumerName = propertyConfigurer.getProperty("activemq.VirtualTopic.consumer.name");
                        if(StringUtil.isBlank(consumerName)) {
                            throw new IllegalArgumentException("请设置activemq.VirtualTopic.consumer.name");
                        }
                    }
                    virtualTopicDestNames.add("Consumer." + consumerName + ".VirtualTopic." + destName);
                }
            }
        }

        mConsume.retainAll(mProduce);
        if(!mConsume.isEmpty()) {
            throw new IllegalArgumentException("生产和消费的分布式事件类型具有交集，这将产生死循环");
        }


        regist();
    }

    /**
     * 判断本程序是否生产此分布式事件类型
     * @param type
     * @return
     */
    public boolean exist(String type) {
        return produce.containsKey(type);
    }

    public Class get(String type) {
        return consume.get(type);
    }

    private void regist() {

        Set<String> destNames = new HashSet<>();
        destNames.addAll(queueDestNames);
        destNames.addAll(virtualTopicDestNames);
        destNames.addAll(topicDestNames);
        if(!destNames.isEmpty()) {
            ConfigurableApplicationContext applicationContext = (ConfigurableApplicationContext) this.applicationContext;
            DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) applicationContext.getBeanFactory();

            BeanDefinitionBuilder distMessageReceiverBuilder = BeanDefinitionBuilder.genericBeanDefinition(DistMessageReceiver.class);
            distMessageReceiverBuilder.setLazyInit(false);
            defaultListableBeanFactory.registerBeanDefinition("distMessageReceiver", distMessageReceiverBuilder.getRawBeanDefinition());

            BeanDefinitionBuilder containerBuilder = BeanDefinitionBuilder.genericBeanDefinition(DefaultMessageListenerContainer.class);
            containerBuilder.setLazyInit(false);

            String distNameStr = CollectionUtil.array2String(new ArrayList<>(destNames));

            containerBuilder.addPropertyValue("concurrentConsumers", 10);
            containerBuilder.addPropertyReference("connectionFactory", "jmsFactory");
            containerBuilder.addPropertyValue("destinationName", distNameStr);
            containerBuilder.addPropertyValue("pubSubNoLocal", false);
            containerBuilder.addPropertyReference("messageListener", "distMessageReceiver");

            defaultListableBeanFactory.registerBeanDefinition("defaultMessageListenerContainer", containerBuilder.getRawBeanDefinition());
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
